package com.patsnap.data.npd.dw.etl.base;

import com.patsnap.one.etl.cdc.CdcRecord;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;

public class RateLimitWatermarkGeneratorSupplier implements WatermarkGeneratorSupplier<CdcRecord> {
    private final long maxRecordsPerSecond;
    private final long watermarkDelayMillis;

    public RateLimitWatermarkGeneratorSupplier(long maxRecordsPerSecond, long watermarkDelayMillis) {
        this.maxRecordsPerSecond = maxRecordsPerSecond;
        this.watermarkDelayMillis = watermarkDelayMillis;
    }

    @Override
    public WatermarkGenerator<CdcRecord> createWatermarkGenerator(Context context) {
        return new RateLimitWatermarkGenerator(maxRecordsPerSecond, watermarkDelayMillis);
    }

}

